package com.db.concurrency.producerconsumer.executors;

import java.io.File;
import java.io.FileFilter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author Alexander Kotov (kotov.alex.22@gmail.com)
 */
public class ProducerConsumer {
    private CountDownLatch crawlDoneSignal = new CountDownLatch(1);
    private static final int N_CONSUMERS = Runtime.getRuntime().availableProcessors();
    private final List<BlockingDeque<File>> tasksList = new ArrayList<BlockingDeque<File>>();
    private ExecutorService producersExecutor;
    private final ExecutorService consumersExecutor = Executors.newFixedThreadPool(N_CONSUMERS);

    public void startIndexing(File root) {
        File[] files = root.listFiles();
        producersExecutor = Executors.newFixedThreadPool(files.length);

        for (int i = 0; i < N_CONSUMERS; ++i) {
            tasksList.add(new LinkedBlockingDeque<File>());
        }

        FileFilter filter = new FileFilter() {
            public boolean accept(File file) {
                return true;
            }
        };

        producersExecutor.submit(new FileCrawler(tasksList, 0, producersExecutor, filter, root, crawlDoneSignal));

        for (int i = 0; i < N_CONSUMERS; i++) {
            consumersExecutor.submit(new FileIndexer(i, tasksList, consumersExecutor));
        }
    }

    public void awaitCrawling() throws InterruptedException {
        System.out.println("waiting...");
        crawlDoneSignal.await();
        System.out.println("crawling finished.");
    }

    public void stopIndexing() throws InterruptedException{
        producersExecutor.shutdown();
        consumersExecutor.shutdownNow();
        consumersExecutor.awaitTermination(10, TimeUnit.MINUTES);
    }

    public static void main(String[] args) throws  InterruptedException {
        File file = new File(args[0]);
        ProducerConsumer producerConsumer = new ProducerConsumer();
        producerConsumer.startIndexing(file);

        producerConsumer.awaitCrawling();

        producerConsumer.stopIndexing();

    }
}
